package com.rem.flink.flink8State;

import com.rem.flink.flink2Source.ClickSource;
import com.rem.flink.flink2Source.Event;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.state.MapState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

import java.sql.Timestamp;

/**
 * 按键分区状态 MapState
 * 模拟滚动窗口  10s/次
 *
 * @author Rem
 * @date 2022-10-25
 */

public class MapStateTest {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        SingleOutputStreamOperator<Event> stream = env.addSource(new ClickSource())
                .assignTimestampsAndWatermarks(WatermarkStrategy.<Event>forMonotonousTimestamps()
                        .withTimestampAssigner((element, recordTimestamp) -> element.getTimestamp()));

        stream.printToErr("time");

        stream.keyBy(Event::getUrl)
                .process(new KeyedProcessFunction<String, Event, String>() {
                    // 定义属性，窗口长度 10s
                    final long windowSize = 10000L;

                    // 声明状态，用map保存pv值（窗口start，count）
                    MapState<Long, Long> windowPvMapState;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        windowPvMapState = getRuntimeContext().getMapState(new MapStateDescriptor<>("window-pv", Long.class, Long.class));
                    }


                    @Override
                    public void processElement(Event value, KeyedProcessFunction<String, Event, String>.Context ctx, Collector<String> out) throws Exception {
                        // 每来一条数据，就根据时间戳判断属于哪个窗口
                        Long windowStart = value.getTimestamp() / windowSize * windowSize;
                        long windowEnd = windowStart + windowSize;

                        ctx.timerService().registerEventTimeTimer(windowEnd - 1);

                        if (windowPvMapState.contains(windowStart)) {
                            windowPvMapState.put(windowStart, windowPvMapState.get(windowStart) + 1);
                        } else {
                            windowPvMapState.put(windowStart, 1L);
                        }
                    }


                    @Override
                    public void onTimer(long timestamp, KeyedProcessFunction<String, Event, String>.OnTimerContext ctx, Collector<String> out) throws Exception {
                        long windowEnd = timestamp + 1;
                        long windowStart = windowEnd - windowSize;
                        Long pv = windowPvMapState.get(windowStart);
                        out.collect("url: " + ctx.getCurrentKey() + " 访问量: " + pv + " 窗口：" + new Timestamp(windowStart) + " ~ " + new Timestamp(windowEnd));

                        // 模拟窗口的销毁，清除map中的key
                        windowPvMapState.remove(windowStart);
                    }
                })
                .print();

        env.execute();


    }
}
